package com.shujia.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo3ReadMysql {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(2)

    //使用自定义source读取mysql
    val mysqlDS: DataStream[(String, String)] = env.addSource(new MysqlSource)

    mysqlDS
      .map(kv => (kv._2, 1))
      .keyBy(_._1)
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()

    env.execute()

  }

}

/**
  *
  * SourceFunction: 单并行度的source< run方法只执行一次
  * RichSourceFunction: 多了open和close方法
  * ParallelSourceFunction: 并行的source， 每一个并行度中run方法都会执行一次
  * RichParallelSourceFunction: 多了open和close方法
  *
  */


class MysqlSource extends RichParallelSourceFunction[(String, String)] {

  /**
    * open方法在run 方法之前执行，一般用于初始化
    *
    */

  var con: Connection = _

  override def open(parameters: Configuration): Unit = {

    println("创建jdbc链接")

    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //2、创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/tour", "root", "123456")

  }

  /**
    * close 方法在run之后执行， 一般用于关闭链接
    *
    */
  override def close(): Unit = {

    con.close()
  }

  override def run(ctx: SourceFunction.SourceContext[(String, String)]): Unit = {

    //3、查询数据
    val stat: PreparedStatement = con.prepareStatement("select city_id,city_name from admin_code")

    //4、执行查询
    val resultSet: ResultSet = stat.executeQuery()

    //5、解析数据
    while (resultSet.next()) {

      val cityId: String = resultSet.getString("city_id")
      val cityName: String = resultSet.getString("city_name")


      //将数据发送到下游
      ctx.collect((cityId, cityName))

    }
  }

  override def cancel(): Unit = {

  }
}